package com.youlin.config;

import com.google.common.collect.Maps;
import org.springframework.amqp.core.*;
import org.springframework.amqp.rabbit.config.SimpleRabbitListenerContainerFactory;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.transaction.RabbitTransactionManager;
import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.DependsOn;

import java.util.HashMap;
import java.util.Map;

//@Configuration
public class RabbitMQConfig {

    //    @Value("${spring.rabbitmq.host}")
    private String host;
    //    @Value("${spring.rabbitmq.port}")
    private int port;
    //    @Value("${spring.rabbitmq.username}")
    private String username;
    //    @Value("${spring.rabbitmq.password}")
    private String password;

    //    @Bean
    public CachingConnectionFactory cachingConnectionFactory() {
        CachingConnectionFactory connectionFactory = new CachingConnectionFactory(host, port);
//        CachingConnectionFactory connectionFactory = new CachingConnectionFactory();
//        connectionFactory.setAddresses("192.168.182.121:5672,192.168.182.122:5672");
        connectionFactory.setUsername(username);
        connectionFactory.setPassword(password);
        connectionFactory.setVirtualHost("/");
//        connectionFactory.setVirtualHost("123");
        connectionFactory.setPublisherConfirms(true);
        connectionFactory.setPublisherReturns(true);
        return connectionFactory;
    }

    //    @Bean
//    @DependsOn("cachingConnectionFactory")
    public RabbitTemplate rabbitTemplate(CachingConnectionFactory cachingConnectionFactory) {
        RabbitTemplate rabbitTemplate = new RabbitTemplate(cachingConnectionFactory);
        rabbitTemplate.setMessageConverter(new Jackson2JsonMessageConverter());
        rabbitTemplate.setMandatory(true);
        // 和 publisherConfirms 不可共存
//        rabbitTemplate.setChannelTransacted(true);
        rabbitTemplate.setConfirmCallback((correlationData, ack, cause) -> {
            if (ack) System.out.printf("\033[32m 消息发送成功抵达服务器，成功确认 ===>> correlationData(%s), " +
                    "ack(%s), cause(%s) \033[0m\n", correlationData, ack, cause);
            else System.out.printf("\033[31m 消息发送未成功抵达服务器，失败确认 ===>> correlationData(%s), " +
                    "ack(%s), cause(%s) \033[0m\n", correlationData, ack, cause);
        });
        rabbitTemplate.setReturnCallback((message, replyCode, replyText, exchange, routingKey) ->
                System.out.printf("\033[31m 消息发送未成功抵达消息队列，被退回给发送者 ===>> exchange(%s), route(%s), replyCode(%s), " +
                        "replyText(%s), message(%s) \033[0m\n", exchange, routingKey, replyCode, replyText, message));
        return rabbitTemplate;
    }

    //    @Bean
//    @DependsOn("cachingConnectionFactory")
    public RabbitTransactionManager rabbitTransactionManager(CachingConnectionFactory cachingConnectionFactory) {
        return new RabbitTransactionManager(cachingConnectionFactory);
    }

    /**
     * 针对消费者配置
     * 1. 设置交换机类型
     * 2. 将队列绑定到交换机
     * FanoutExchange：将消息分发到所有的绑定队列，无routingkey的概念
     * HeadersExchange：通过添加属性key-value匹配
     * DirectExchange：按照routingkey分发到指定队列
     * TopicExchange：多关键字匹配
     *
     * @return
     */
    @Bean
    public DirectExchange directExchangeA() {
        return new DirectExchange("DIRECTEXCHANGE_A");
    }

    @Bean
    public DirectExchange directExchangeB() {
        return new DirectExchange("DIRECTEXCHANGE_B");
    }

    @Bean
    public FanoutExchange fanoutExchange() {
        return new FanoutExchange("FANOUTEXCHANGE");
    }

    @Bean
    public FanoutExchange fanoutExchangeA() {
        return new FanoutExchange("FANOUTEXCHANGE_A");
    }

    @Bean
    public FanoutExchange fanoutExchangeB() {
        return new FanoutExchange("FANOUTEXCHANGE_B");
    }

    @Bean
    public TopicExchange topicExchange() {
        return new TopicExchange("TOPICEXCHANGE");
    }

    @Bean
    public HeadersExchange headersExchange() {
        return new HeadersExchange("HEADERSEXCHANGE");
    }

    @Bean
    public Queue queue0() {
        Map<String, Object> params = new HashMap<>();
        // x-message-ttl 声明了队列里的消息统一的过期时间
//        params.put("x-message-ttl", 5 * 1000);
        // x-dead-letter-exchange 声明了队列里的死信转发到的DLX名称，4
        params.put("x-dead-letter-exchange", "DIRECTEXCHANGE_B");
        // x-dead-letter-routing-key 声明了这些死信在转发时携带的 routing-key 名称。
        params.put("x-dead-letter-routing-key", "routingKey_B_B");
        return new Queue("QUEUE_0", true, false, false, params); //队列持久
    }

    @Bean
    public Queue queueA() {
        return new Queue("QUEUE_A", true); //队列持久
    }

    @Bean
    public Queue queueB() {
        return new Queue("QUEUE_B", true); //队列持久
    }

    // 一个交换机可以绑定多个消息队列，也就是消息通过一个交换机，可以分发到不同的队列当中去
    @Bean
    public Binding bindingDA2QA(Queue queueA, DirectExchange directExchangeA) {
        return BindingBuilder.bind(queueA).to(directExchangeA).with("routingKey_A_A");
    }

    @Bean
    public Binding bindingDA2Q0(Queue queue0, DirectExchange directExchangeA) {
        return BindingBuilder.bind(queue0).to(directExchangeA).with("routingKey_0_A");
    }

    @Bean
    public Binding bindingDB2QA(Queue queueA, DirectExchange directExchangeB) {
        return BindingBuilder.bind(queueA).to(directExchangeB).with("routingKey_A_B");
    }

    @Bean
    public Binding bindingDB2Q0(Queue queue0, DirectExchange directExchangeB) {
        return BindingBuilder.bind(queue0).to(directExchangeB).with("routingKey_0_B");
    }

    @Bean
    public Binding bindingDA2QB(Queue queueB, DirectExchange directExchangeA) {
        return BindingBuilder.bind(queueB).to(directExchangeA).with("routingKey_B_A");
    }

    @Bean
    public Binding bindingDB2QB(Queue queueB, DirectExchange directExchangeB) {
        return BindingBuilder.bind(queueB).to(directExchangeB).with("routingKey_B_B");
    }

    @Bean
    public Binding bindingFA2F(FanoutExchange fanoutExchange, FanoutExchange fanoutExchangeA) {
        return BindingBuilder.bind(fanoutExchange).to(fanoutExchangeA);
    }

    @Bean
    public Binding bindingFB2F(FanoutExchange fanoutExchange, FanoutExchange fanoutExchangeB) {
        return BindingBuilder.bind(fanoutExchange).to(fanoutExchangeB);
    }

    @Bean
    public Binding bindingF2Q0(Queue queue0, FanoutExchange fanoutExchange) {
        return BindingBuilder.bind(queue0).to(fanoutExchange);
    }

    @Bean
    public Binding bindingF2QA(Queue queueA, FanoutExchange fanoutExchangeA) {
        return BindingBuilder.bind(queueA).to(fanoutExchangeA);
    }

    @Bean
    public Binding bindingF2QB(Queue queueB, FanoutExchange fanoutExchangeB) {
        return BindingBuilder.bind(queueB).to(fanoutExchangeB);
    }

    @Bean
    public Binding bindingT2Q0(Queue queue0, TopicExchange topicExchange) {
        return BindingBuilder.bind(queue0).to(topicExchange).with("0.*");
    }

    @Bean
    public Binding bindingT2QA(Queue queueA, TopicExchange topicExchange) {
        return BindingBuilder.bind(queueA).to(topicExchange).with("A.*");
    }

    @Bean
    public Binding bindingT2QB(Queue queueB, TopicExchange topicExchange) {
        return BindingBuilder.bind(queueB).to(topicExchange).with("B.#");
    }

    @Bean
    public Binding bindingH2QB(Queue queueB, HeadersExchange headersExchange) {
//        return BindingBuilder.bind(queueB).to(headersExchange).where("vip").exists();
//        return BindingBuilder.bind(queueB).to(headersExchange).where("vip").matches(true);
//        return BindingBuilder.bind(queueB).to(headersExchange).whereAny("vip", "grade").exist();
//        return BindingBuilder.bind(queueB).to(headersExchange).whereAll("vip", "grade").exist();
        HashMap<String, Object> header = Maps.newHashMap();
        header.put("vip", true);
        header.put("grade", 12);
//        return BindingBuilder.bind(queueB).to(headersExchange).whereAny(header).match();
        return BindingBuilder.bind(queueB).to(headersExchange).whereAll(header).match();
    }

    @Bean // 单线程消费者
    @DependsOn("cachingConnectionFactory")
    public SimpleRabbitListenerContainerFactory singleListenerContainer(CachingConnectionFactory cachingConnectionFactory) {
        SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
        factory.setConnectionFactory(cachingConnectionFactory);
        factory.setMessageConverter(new Jackson2JsonMessageConverter());
        factory.setConcurrentConsumers(1);
        factory.setMaxConcurrentConsumers(1);
//        factory.setPrefetchCount(5); // 分发机制-轮询(根据prefetchCount，例如A5-B5-A5-B5，不考虑AB的消费能力)
        factory.setPrefetchCount(1); // 分发机制-公平(根据消费能力，多劳多得)
        factory.setTxSize(1);
//        factory.setAcknowledgeMode(AcknowledgeMode.AUTO);
//        factory.setAcknowledgeMode(AcknowledgeMode.MANUAL);
        factory.setChannelTransacted(true);
        return factory;
    }

    @Bean // 多线程消费者
    @DependsOn("cachingConnectionFactory")
    public SimpleRabbitListenerContainerFactory multiListenerContainer(CachingConnectionFactory cachingConnectionFactory) {
        SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
        factory.setConnectionFactory(cachingConnectionFactory);
        factory.setMessageConverter(new Jackson2JsonMessageConverter());
        factory.setConcurrentConsumers(5);
        factory.setMaxConcurrentConsumers(10);
        factory.setPrefetchCount(3);
        factory.setAcknowledgeMode(AcknowledgeMode.MANUAL);
        return factory;
    }

}
